package cn.itcast.model.matchtag

import java.util.Properties

import cn.itcast.model.bean.{HBaseMeta, TagRule}
import org.apache.spark.SparkContext
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object GenderModel {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("GenderModel")
      .master("local[*]")
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    val url = "jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val table = "tbl_basic_tag"
    val properties = new Properties()
    val mysqlDF: DataFrame = spark.read.jdbc(url, table, properties)
    val fourDS: Dataset[Row] = mysqlDF.select("id","rule").where("id = 606")
    import spark.implicits._
    val meta: HBaseMeta = fourDS.map(row => {
      val ruleSource: String = row.getAs[String]("rule")
      val tmpArr: Array[String] = ruleSource.split("##")
      val tuples: Array[(String, String)] = tmpArr.map(kv => {
        val arr: Array[String] = kv.split("=")
        (arr(0), arr(1))
      })
      val map: Map[String, String] = tuples.toMap
      HBaseMeta(map)
    }).collect()(0)
    val fiveDS: Dataset[Row] = mysqlDF.select("id","rule").where("pid = 606")
    val fiveRuleList: List[TagRule] = fiveDS.map(row => {
      val id: String = row.getAs[Int]("id").toString
      val rule: String = row.getAs[Int]("rule").toString
      TagRule(id, rule)
    }).collect().toList
    val hbaseSource: DataFrame = spark.read
      .format("cn.itcast.up.model.HBaseSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, meta.hbaseTable)
      .option(HBaseMeta.FAMILY, meta.family)
      .option(HBaseMeta.SELECTFIELDS, meta.selectFields)
      .load()
    import org.apache.spark.sql.functions._
    val matchTag: UserDefinedFunction = udf((gender: String) => {
      var tagId = ""
      for (tag <- fiveRuleList) {
        if (tag.rule == gender) {
          tagId = tag.id
        }
      }
      tagId
    })
    val result: DataFrame = hbaseSource.select('id.as("userId"),matchTag('gender).as("tagIds"))
    result.write
      .format("cn.itcast.model.util.HbaseSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test34")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagIds")
      .save()
  }
}
